package cn.itcast.cep;

import cn.itcast.bean.Product;
import cn.itcast.config.QuotConfig;
import cn.itcast.util.RedisUtil;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternSelectFunction;
import org.apache.flink.cep.PatternStream;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer011;
import org.apache.flink.util.Collector;
import redis.clients.jedis.JedisCluster;

import java.util.List;
import java.util.Map;
import java.util.Properties;
/*
{"goodsId":100001,"goodsPrice":6,"goodsName":"apple","alias":"苹果","orderTime":1558430843000}
{"goodsId":100007,"goodsPrice":0.5,"goodsName":"mask","alias":"口罩","orderTime":1558430844000}
{"goodsId":100002,"goodsPrice":2,"goodsName":"rice","alias":"大米","orderTime":1558430845000}
{"goodsId":100003,"goodsPrice":2,"goodsName":"flour","alias":"面粉","orderTime":1558430846000}
{"goodsId":100004,"goodsPrice":12,"goodsName":"rice","alias":"大米","orderTime":1558430847000}
{"goodsId":100005,"goodsPrice":20,"goodsName":"apple","alias":"苹果","orderTime":1558430848000}
{"goodsId":100006,"goodsPrice":3,"goodsName":"banana","alias":"香蕉","orderTime":1558430849000}
{"goodsId":100007,"goodsPrice":10,"goodsName":"mask","alias":"口罩","orderTime":1558430850000}
{"goodsId":100001,"goodsPrice":16,"goodsName":"apple","alias":"苹果","orderTime":1558430852000}
{"goodsId":100007,"goodsPrice":15,"goodsName":"mask","alias":"口罩","orderTime":1558430853000}
{"goodsId":100002,"goodsPrice":12,"goodsName":"rice","alias":"大米","orderTime":1558430854000}
{"goodsId":100003,"goodsPrice":12,"goodsName":"flour","alias":"面粉","orderTime":1558430855000}
{"goodsId":100004,"goodsPrice":12,"goodsName":"rice","alias":"大米","orderTime":1558430856000}
{"goodsId":100005,"goodsPrice":20,"goodsName":"apple","alias":"苹果","orderTime":1558430857000}
{"goodsId":100006,"goodsPrice":13,"goodsName":"banana","alias":"香蕉","orderTime":1558430858000}
{"goodsId":100007,"goodsPrice":10,"goodsName":"mask","alias":"口罩","orderTime":1558430859000}
*/

public class CepMarkets {
    public static void main(String[] args) throws Exception {
        /**
         * 1.获取流处理执行环境
         * 2.设置事件时间、并行度
         * 3.整合kafka
         * 4.数据转换
         * 5.process获取bean,设置status，并设置事件时间
         * 6.定义匹配模式，设置时间长度
         * 7.匹配模式（分组）
         * 8.查询告警数据
         */

        //todo 1.获取流处理执行环境
        //todo 2.设置事件时间、并行度

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

        //todo 3.整合kafka
        Properties prop = new Properties();
        prop.setProperty("bootstrap.servers", QuotConfig.config.getProperty("bootstrap.servers"));
        FlinkKafkaConsumer011<String> kafkaSource = new FlinkKafkaConsumer011<>("cep", new SimpleStringSchema(), prop);
        kafkaSource.setStartFromEarliest();
        //todo 4.数据转换
        SingleOutputStreamOperator<Product> mapData = env.addSource(kafkaSource).map(new MapFunction<String, Product>() {
            @Override
            public Product map(String s) throws Exception {
                JSONObject jsonObject = JSON.parseObject(s);
                Product product = new Product(
                        jsonObject.getLong("goodsId"),
                        jsonObject.getDouble("goodsPrice"),
                        jsonObject.getString("goodsName"),
                        jsonObject.getString("alias"),
                        jsonObject.getLong("orderTime"),
                        false
                );
                return product;
            }
        });

        //todo 5.process获取bean,设置status，并设置事件时间
        //比较商品价格和阈值数据，设置status
        SingleOutputStreamOperator<Product> waterData = mapData.process(new ProcessFunction<Product, Product>() {
            JedisCluster jedisCluster = null;

            @Override
            public void open(Configuration parameters) throws Exception {
                //获取redis链接对象
                jedisCluster = RedisUtil.getJedisCluster();
            }

            @Override
            public void processElement(Product product, ProcessFunction<Product, Product>.Context context, Collector<Product> collector) throws Exception {
                //阈值数据
                String threshold = jedisCluster.hget("product", product.getGoodsName());
                if (product.getGoodsPrice() > Double.valueOf(threshold)) {
                    product.setStatus(true);
                }
                collector.collect(product);
            }
        }).assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<Product>(Time.seconds(0)) {
            @Override
            public long extractTimestamp(Product product) {
                return product.getOrderTime();
            }
        });

        //todo 6.定义匹配模式，设置时间长度
        //商品售价在1分钟之内有连续两次超过预定商品价格阀值就发送告警信息。
        Pattern<Product, Product> pattern = Pattern.<Product>begin("begin")
                .where(new SimpleCondition<Product>() {
                    @Override
                    public boolean filter(Product product) throws Exception {
                        return product.getStatus()==true;
                    }
                })
                .next("next")
                .where(new SimpleCondition<Product>() {
                    @Override
                    public boolean filter(Product product) throws Exception {
                        return product.getStatus() == true;
                    }
                }).within(Time.seconds(60));

        //todo 7.匹配模式（分组）
        PatternStream<Product> cep = CEP.pattern(waterData.keyBy(Product::getGoodsId), pattern);
        //todo 8.查询告警数据
        cep.select(new PatternSelectFunction<Product, Object>() {
            @Override
            public Object select(Map<String, List<Product>> map) throws Exception {
                return map.get("next");
            }
        }).print();

        env.execute();
    }
}

